package com.zero.java.base.stream;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author Lenovo 并行流处理
 */
public class ParallelStreamAPI {

	private final static String file_name = "cfg/words.txt";

	public static void main(String[] args) throws IOException {

		String contents = new String(Files.readAllBytes(Paths.get(file_name)), StandardCharsets.UTF_8);
		List<String> words = Arrays.asList(contents.split("\\PL+"));
		// 对字符传流中的所有短单词计数
		// 问题：传递给forEach的函数会在多个线程中运行，每一个都会更新共享数组，所以运行多次，很可能发现每次运行都会产生不同的计数值，而且每个都是错的；
		int[] shortWords = new int[6];
		words.parallelStream().forEach(s -> {
			if (s.length() < 6)
				shortWords[s.length()]++;
		});
		System.out.println("shortWords:" + Arrays.toString(shortWords));

		// 处理以上对应问题：要保证传递给并行流操作的任何函数都可以安全的并行的执行，达到这个目的的最佳方法是远离移变状态；
		Stream<String> parallelStream = Stream.of(contents.split("\\PL+")).parallel();
		Map<Integer, Long> collect = parallelStream.filter(s -> s.length() > 6)
				.collect(Collectors.groupingBy(String::length, Collectors.counting()));
		System.out.println("collect:" + collect);

		// parallelStream
		Stream<String> parallelStream2 = words.parallelStream();
		long wordsCount = parallelStream2.filter(s -> s.length() > 6).count();
		System.out.println("wordsCount:" + wordsCount);

		// 合并映射表的代价很昂贵，groupingByConcurrent方法使用了共享的并发映射表
		ConcurrentMap<Integer, List<String>> concurrentMap = words.parallelStream()
				.collect(Collectors.groupingByConcurrent(String::length));
		System.out.println(concurrentMap.get(6));

		ConcurrentMap<Integer, Long> concurrentMap2 = words.parallelStream()
				.collect(Collectors.groupingByConcurrent(String::length, Collectors.counting()));
		System.out.println(concurrentMap2);
		System.out.println(concurrentMap2.get(6));

	}

}
